package org.example.rabbitmq.client.concurrent;

import com.rabbitmq.client.Channel;
import org.example.rabbitmq.config.LatencyProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class LatencyChannelAwareMessageListener implements ChannelAwareMessageListener {
    Logger logger = LoggerFactory.getLogger(LatencyChannelAwareMessageListener.class);
    @Autowired
    LatencyProperties latencyProperties;
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Override
    public void onMessage(Message message, Channel channel) {
        String currentQueue = message.getMessageProperties().getConsumerQueue();
        logger.info("[C] invoking remote service..., in " + currentQueue);
        try {
            invokeRemoteService();
        } catch (TimeoutException e) {
            rejectMessage(message, channel);
            latencyPublish(message);
        } catch (Exception e) {
            rejectMessage(message, channel);
        }
    }

    /**
     * 当异常时，拒绝此消息，并且不重新入队
     *
     * @param message
     * @param channel
     */
    private void rejectMessage(Message message, Channel channel) {
        String currentQueue = message.getMessageProperties().getConsumerQueue();
        logger.info("[C] invoking remote service timeout, in " + currentQueue);
        try {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException ex) {
            logger.error("[C] reject failed", ex);
        }
    }

    /**
     * 重新发布消息到延迟队列
     *
     * @param message
     */
    private void latencyPublish(Message message) {
        String currentQueue = message.getMessageProperties().getConsumerQueue();
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        try {
            final String latencyExchangeName = latencyProperties.getExchangeName();

            MessageProperties messageProperties = new MessageProperties();
            // 重置过期时间为2分钟，最终过期时间取较小值，即：MIN(队列的x-message-ttl属性值, 该值)
            messageProperties.setExpiration("120000");
            // messageProperties.setReceivedRoutingKey(routingKey);
            Message latencyMessage = new Message(message.getBody(), messageProperties);
            rabbitTemplate.convertAndSend(latencyExchangeName, routingKey, latencyMessage);
            logger.info("[C] the message requeue to latency queue, in " + currentQueue + " with routingKey " + routingKey);
        } catch (Exception ex) {
            logger.error("[C] requeue to latency queue failed", ex);
        }
    }

    private void invokeRemoteService() throws TimeoutException {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        throw new TimeoutException("同步数据超时");
    }
}
